package com.atguigu.gmall.realtime.demo;

import org.apache.doris.flink.cfg.DorisStreamOptions;
import org.apache.doris.flink.datastream.DorisSourceFunction;
import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @author caodan
 * @version 1.0
 * @description flink 操作doris
 * @date 2025-08-01 14:48
 */
public class FlinkDorisReadDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.setParallelism(1);
        Properties props = new Properties();
        props.setProperty("fenodes", "192.168.10.102:8030");
//        props.setProperty("benodes", "192.168.10.102:9060");
        props.setProperty("username", "root");
        props.setProperty("password", "Caodan520@");
        props.setProperty("table.identifier", "doris_test.hello");
        props.setProperty("stream-load.enable", "true");
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        env
                .addSource(new DorisSourceFunction(new DorisStreamOptions(props), new SimpleListDeserializationSchema()))
                .print();

        env.execute();
    }
}
